package shotgun.my.sweetutil;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;

/**
 * 多线程异步编程工具类
 *
 * @author wulingming
 **/
public class StreamUtils {

    /**cpu线程数*/
    public static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();

    /**
     * io密集型任务可以多分配点线程
     * 这里不使用AVAILABLE_PROCESSORS - 1，是因为只有stream原始方式，main主线程才参与forkjoin。所以这里总线程数即为cpu核心线程数
     */
    public static final int DEFAULT_PARALLELISM = AVAILABLE_PROCESSORS <= 0 ? 1 : AVAILABLE_PROCESSORS;

    /**io密集型线程池*/
    public static final ForkJoinPool FORK_JOIN_POOL_IO = new ForkJoinPool(DEFAULT_PARALLELISM);
    public static final ForkJoinPool FORK_JOIN_POOL_OTHEN = new ForkJoinPool(DEFAULT_PARALLELISM);

    /**
     * 执行任务(主线程 等待 任务完成)
     * 不使用默认的线程池,避免导致整个项目卡顿{@link ForkJoinPool#commonPool()}
     * 适用于IO密集型任务，比如：io流操作、http请求等。
     *
     * @param runnable 执行任务
     **/
    public static void run(Runnable runnable) {
        execute(FORK_JOIN_POOL_OTHEN, runnable, false);
    }

    /**
     * 执行任务(主线程 等待 任务完成)
     * 不使用默认的线程池,避免导致整个项目卡顿{@link ForkJoinPool#commonPool()}
     * 适用于IO密集型任务，比如：io流操作、http请求等。
     *
     * @param forkJoinPool 线程池
     * @param runnable     执行任务
     **/
    public static void run(ForkJoinPool forkJoinPool, Runnable runnable) {
        execute(forkJoinPool, runnable, false);
        //不关闭ForkJoinPool，由外部自己管理
    }

    /**
     * 执行任务(主线程 不等待 任务完成)
     * 不使用默认的线程池,避免导致整个项目卡顿{@link ForkJoinPool#commonPool()}
     * 适用于IO密集型任务，比如：io流操作、http请求等。
     *
     * @param runnable 执行任务
     **/
    public static void runAsync(Runnable runnable) {
        execute(FORK_JOIN_POOL_OTHEN, runnable, true);
    }

    /**
     * 执行任务(主线程 不等待 任务完成)
     * 不使用默认的线程池,避免导致整个项目卡顿{@link ForkJoinPool#commonPool()}
     * 适用于IO密集型任务，比如：io流操作、http请求等。
     *
     * @param forkJoinPool 线程池
     * @param runnable     执行任务
     **/
    public static void runAsync(ForkJoinPool forkJoinPool, Runnable runnable) {
        execute(forkJoinPool, runnable, true);
    }

    /**
     * 使用jdk默认的线程池执行任务(主线程 不等待 任务完成)
     * 不适用于IO密集型任务，比如：io流操作、http请求等。
     * 该方式总线程数比cpu核心线程数少1，是因为主线程不参与执行。原始stream并行下，主线程参与并行
     *
     * @param runnable 执行任务
     **/
    public static void runAsyncByCommonPool(Runnable runnable) {
        execute(ForkJoinPool.commonPool(), runnable, true);
    }


    /**
     * 执行
     *
     * @param forkJoinPool 线程池
     * @param runnable     执行任务
     * @param isAsync      是否异步，true：异步，false：同步（等待任务完成）
     **/
    private static void execute(ForkJoinPool forkJoinPool, Runnable runnable, boolean isAsync) {
        if (forkJoinPool==null){
            throw new IllegalArgumentException("forkJoinPool不能为空!!!");
        }
        if (runnable==null){
            throw new IllegalArgumentException("runnable不能为空!!!");
        }

        ForkJoinTask<?> forkJoinTask = forkJoinPool.submit(runnable);
        if (!isAsync) {
            //等待子线程任务完成
            forkJoinTask.join();
        }
    }

//    public static void main(String[] args) {
//        final Integer[] result = new Integer[1];
//        String[] firstRange = new String[50];
//
//        runAsyncByCommonPool(() -> {
//            //累加总次数
//            Integer r = Stream.of(firstRange).parallel().map((number) -> {
//                System.out.println("线程:" + Thread.currentThread().getName());
//                return 1;
//            }).reduce(Integer::sum).orElse(0);
//
//            result[0] = r;
//
//            System.out.println("count:" + r);
//
//        });
//
//        try {
//            Thread.sleep(3000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println("并行计算结果:" + result[0]);
//
//        System.out.println("ok!!!!!!!!!!!!!!!!!!!");
//    }



    public static void main2(String[] args) throws ExecutionException, InterruptedException {
        // 没有返回值
        CompletableFuture<Void> runAsync =
                CompletableFuture.runAsync(() -> System.out.println("已执行runAsync 200"));
        runAsync.get(); // 获取结果值，如果一直执行不完，该方法就会被阻塞，一直等待去get

        // 有返回值
        CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println("准备等待");
            // System.out.println(19/0);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 200;
        });
        //System.out.println(supplyAsync.get());


        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 当成功完成
        Integer info = supplyAsync.whenComplete((t, e) -> {
            System.out.println(t); // 获得成功执行完成的返回值，如果执行出错为null
            System.out.println(e); // 如果执行不成功，获取异常信息，如果没有异常为null
        }).exceptionally((e) -> { // 如果没有异常不执行
            e.getMessage(); // 将执行失败的异常信息获取出来
            return 404; // 有返回值
        }).get();

        System.out.println(info);

    }
}
